package HHD.Sgg.UserADS

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/**
 * 龚英杰：
 * 找到用户每天第一次访问的页面、倒数第二次访问的页面和访问最多的页面
 * 将结果存入 HHDtest.ads_user_FirstAndSecondToLastAndMaxPage
 * 字段           类型          说明
 * user_id      string        用户id
 * user_name    String        用户姓名
 * first_page   String        第一次访问的页面名
 * s_to_last    String        倒数第二次访问的页面名(如果只有一个访问，则返回该访问）
 * max_the_page String        访问最多的页面名(如果有多个最多，则返回一个集合，使用 ,分隔）
 * dt         String        日期(分区列）
 */


object UserLastPage {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().getOrCreate()

    spark.sql(
      """
        |insert into table HHDtest.ads_user_FirstAndSecondToLastAndMaxPage
        |(with t1 as(from HHDtest.fact_page_residence_time r join HHDtest.dim_page_info p on r.page_id = p.id join HHDtest.dim_user_info u on r.user_id = u.id
        |   select
        |       user_id,
        |       u.name user_name,
        |       page_id,
        |       p.page,
        |       row_number() over(partition by user_id order by go_in_time asc) r,
        |       max(row_number() over(partition by user_id order by go_in_time asc)) over(partition by user_id) maxr,
        |       count(page_id) over(partition by user_id,page_id) cnt,
        |       max(count(page_id) over(partition by user_id,page_id)) over(partition by user_id)maxcnt,
        |       date_format(go_in_time,'yyyy-MM-dd') date)
        |from(from((from t1
        |   select
        |       user_id,
        |       user_name,
        |       page,
        |       date
        |   where r = 1)r1
        |join
        |(from t1
        |   select
        |       page,
        |       user_id
        |   where r = if(maxr = 1 , maxr,maxr-1))r2
        |on r1.user_id=r2.user_id
        |join
        |(from t1
        |   select
        |       page,
        |       user_id
        |   where cnt = maxcnt)r3
        |on r1.user_id = r3.user_id)
        |   select
        |   distinct
        |     r1.user_id,
        |     r1.user_name,
        |     r1.page first_page,
        |     r2.page s_to_last,
        |     r3.page max_the_page,
        |     date)g
        |select
        |distinct
        |   user_id,
        |   user_name,
        |   first_page,
        |   s_to_last,
        |   case when count(max_the_page) over(partition by user_id)>=2 then concat_ws(',',collect_list(max_the_page) over(partition by user_id)) else max_the_page end max_the_page,
        |   date dt)
        |""".stripMargin)


  }
}
